# Databricks notebook source
# MAGIC %md
# MAGIC 
# MAGIC # 概述
# MAGIC 
# MAGIC **结构化流（Structured Streaming）**式处理是在Spark SQL引擎上构建的可扩展且容错的流处理引擎。可以像对静态数据表达批处理计算一样表达流式计算。Spark SQL引擎将负责以增量方式连续运行它，并在流数据继续到达时更新最终结果。系统通过检查点和预写日志确保端到端的一次容错保证。
# MAGIC 
# MAGIC > 简而言之，结构化流式处理提供快速、可扩展、容错、端到端的一次性流处理，而无需用户对流进行推理。

# COMMAND ----------

# MAGIC %md
# MAGIC 
# MAGIC ## 入门示例
# MAGIC 
# MAGIC 假设我们希望维护从侦听 TCP 套接字的数据服务器接收的文本数据的运行字数统计。

# COMMAND ----------

# MAGIC %md
# MAGIC 
# MAGIC ### Spark Streaming(DStreams)实现方式
# MAGIC 
# MAGIC 在学习Spark Streaming(DStreams)的时候，我们实现过监听TCP套接字来统计单词个数的功能。

# COMMAND ----------

from pyspark.streaming import StreamingContext

ssc = StreamingContext(sc, 10)

# lines = ssc.socketTextStream("localhost", 5555)
lines = ssc.socketTextStream("104.208.119.0", 5555)

words = lines.flatMap(lambda x: x.split(" ")).map(lambda x: (x, 1)).reduceByKey(lambda a,b: a + b)

words.pprint()

ssc.start()
ssc.awaitTermination()

# COMMAND ----------

ssc.stop()

# COMMAND ----------

# MAGIC %md
# MAGIC 
# MAGIC ### Structured Streaming实现方式
# MAGIC 
# MAGIC 下面我们来看看用Structured Streaming的方式如何实现。

# COMMAND ----------

import pyspark.sql.functions as F

# 获取行数据
lines = spark \
    .readStream \
    .format("socket") \
    .option("host", "104.208.119.0") \
    .option("port", 5555) \
    .load()

print(type(lines))

# 将行拆分成单词
words = lines.select(
   # 炸裂函数
   F.explode(
       F.split(lines.value, " ")
   ).alias("word")
)

print(type(words))

# 统计单词的个数
wordCounts = words.groupBy("word").count()

print(type(wordCounts))

display(wordCounts)

# query = wordCounts \
#     .writeStream \
#     .outputMode("complete") \
#     .format("console")

# print(type(query))

# query.start()
# query.awaitTermination()

# COMMAND ----------

query.stop()

# COMMAND ----------

# MAGIC %md
# MAGIC 
# MAGIC # 编程模型
# MAGIC 
# MAGIC 结构化流中的关键思想是将实时数据流视为连续追加的表。这将导致与批处理模型非常相似的新流处理模型。将流式处理计算表示为标准批处理式查询，就像在静态表上一样，Spark将其作为增量查询在未绑定的输入表上运行。

# COMMAND ----------

# MAGIC %md
# MAGIC 
# MAGIC ## 基本概念
# MAGIC 
# MAGIC 将输入数据流视为“输入表”。到达流的每个数据项都像是追加到输入表的新行。
# MAGIC 
# MAGIC ![](https://spark.apache.org/docs/3.2.1/img/structured-streaming-stream-as-a-table.png)
# MAGIC 
# MAGIC 对输入的查询将生成“结果表”。每个触发间隔（例如，1秒），新行都会追加到输入表中，最终会更新结果表。每当更新结果表时，我们都希望将更改的结果行写入外部接收器。
# MAGIC 
# MAGIC ![](https://spark.apache.org/docs/3.2.1/img/structured-streaming-model.png)
# MAGIC 
# MAGIC “输出”被定义为写出到外部存储的内容。输出可以在不同的模式下定义：
# MAGIC * 完整模式（Complete Mode） - 整个更新的结果表将写入外部存储。由存储连接器决定如何处理整个表的写入。
# MAGIC * 追加模式（Append Mode） - 只有自上次触发器以来在结果表中追加的新行才会写入外部存储。这仅适用于结果表中现有行不应更改的查询。
# MAGIC * 更新模式（Update Mode） - 只有自上次触发器以来在结果表中更新的行才会写入外部存储（自 Spark 2.1.1 起可用）。请注意，这与完整模式的不同之处在于，此模式仅输出自上次触发器以来已更改的行。如果查询不包含聚合，则它将等效于追加模式。
# MAGIC 
# MAGIC ![](https://spark.apache.org/docs/3.2.1/img/structured-streaming-example-model.png)
# MAGIC 
# MAGIC > **结构化流式处理不会具体化整个表。** 它从流数据源读取最新的可用数据，以增量方式处理该数据以更新结果，然后丢弃源数据。它仅保留更新结果所需的最小中间状态数据（例如，前面示例中的中间计数）。

# COMMAND ----------

# MAGIC %md
# MAGIC 
# MAGIC ## 处理事件时间和延迟数据
# MAGIC 
# MAGIC **事件时间**是嵌入在数据本身中的时间，也就是产生增量数据的时间，而不是Spark接收增量数据的时间，这个时间体现在数据中可以作为一列存在，比如数据中存在UPDATED_TS字段表示数据被更新的时间。这就可以支持我们进行基于时间窗口的聚合操作（例如每分钟的事件数量），只要针对数据中的UPDATED_TS字段进行分组和聚合即可。每个时间窗口就是一个分组，而每一行都可以落入对应的分组内。因此，类似这样的基于时间窗口的分组聚合操作，既可以被定义在一份静态数据上，也可以被定义在一个实时数据流上。
# MAGIC 
# MAGIC 此外，这种模型也天然支持延迟到达的数据。Spark会负责更新结果表，因此它有绝对的控制权来针对延迟到达的数据进行聚合结果的重新计算。自Spark2.1之后Structured Streaming开始支持 watermarking(水印)，允许用户指定延时数据的阈值，并允许引擎相应地清除阈值范围之外的旧状态。

# COMMAND ----------

# MAGIC %md
# MAGIC 
# MAGIC ## 容错语义
# MAGIC 
# MAGIC 提供端到端的精确一次语义是Structured Streaming设计背后的关键目标之一。为了实现这一目标，Structured Streaming设计了源（sources）、接收器（sinks）和执行引擎（execution engine），以可靠地跟踪处理的确切进度，以便它可以通过重新启动和/或重新处理来处理任何类型的故障。假定每个流源都有偏移量（类似于 Kafka 偏移量或 Kinesis 序列号）来跟踪流中的读取位置。引擎使用检查点（checkpointing）和预写日志（write-ahead logs）来记录每个触发器中正在处理的数据的偏移范围。接收器设计成可以支持在多次计算处理时保持幂等性。结合使用可重放源和幂等接收器，结构化流式处理可以在任何故障下确保端到端的精确一次语义。

# COMMAND ----------

# MAGIC %md
# MAGIC 
# MAGIC # 使用Datasets和DataFrames的API
# MAGIC 
# MAGIC 从Spark 2.0开始，DataFrame和Dataset可以表示静态的有界数据，也可以表示流式处理、无界数据。与静态DataFrame/Dataset类似，可以使用通用入口点SparkSession从流式处理源创建流式处理DataFrame/Dataset，并对它们应用与静态DataFrame/Dataset相同的操作。

# COMMAND ----------

# MAGIC %md
# MAGIC 
# MAGIC ## 创建流式DataFrames和流式Datasets
# MAGIC 
# MAGIC 流式DataFrame可以通过SparkSession.readStream()返回的DataStreamReader接口进行创建。与用于创建静态DataFrame的read接口类似，可以指定源的详细信息：数据格式（data format）、架构（schema）、选项（options）等。

# COMMAND ----------

# MAGIC %md
# MAGIC 
# MAGIC ### 输入源
# MAGIC 
# MAGIC 有一些内置源。
# MAGIC 
# MAGIC * 文件源 - 读取作为数据流写入目录中的文件。文件将按文件修改时间的顺序进行处理。支持的文件格式包括文本、CSV、JSON、ORC、Parquet。请注意，文件必须以原子方式放置在给定目录中，这在大多数文件系统中可以通过文件移动操作来实现。
# MAGIC 
# MAGIC * Kafka源 - 从Kafka读取数据。它与Kafka版本0.10.0或更高版本兼容。
# MAGIC 
# MAGIC * Socket源（用于测试）- 从Socket连接读取UTF8文本数据。请注意，这应仅用于测试，因为这不提供端到端容错保证。
# MAGIC 
# MAGIC * Rate源（用于测试）- 以每秒指定的行数生成数据，每个输出行包含一个timestamp和value。其中timestamp是包含消息调度时间的Timestamp类型，value是包含消息计数的Long类型，从0开始作为第一行。此源用于测试和基准测试。
# MAGIC 
# MAGIC 某些源不具有容错能力，因为它们不保证在发生故障后可以使用检查点偏移量重放数据，比如Socket源。
# MAGIC 
# MAGIC 以下是 Spark 中所有源的详细信息。
# MAGIC 
# MAGIC | 源           | 选项                                                         | 容错 |
# MAGIC | :----------- | :----------------------------------------------------------- | :--- |
# MAGIC | **文件源**   | `path`：输入目录的路径，并且对所有文件格式通用。<br/><br/>`maxFilesPerTrigger`：每个触发器中要考虑的最大新文件数（默认：无最大值）。<br/><br/>`latestFirst`： 是否首先处理最新的新文件，当存在大量积压文件时很有用（默认：false）。<br/><br/>`fileNameOnly`： 是否仅基于文件名而不是完整路径检查新文件（默认：false）。<br/>将此设置为`true`时，以下文件将被视为同一文件，因为它们的文件名“dataset.txt”是相同的： <br/>"file:///dataset.txt"<br/>"s3：//a/dataset.txt"<br/>"s3n：//a/b/dataset.txt"<br/>"s3a：//a/b/c/dataset.txt"<br/><br/>`maxFileAge`：在此目录中可以找到的文件的最大期限，在这之前的会被忽略。对于第一批，所有文件都将被视为有效。<br/>如果`latestFirst`设置为 `true` 并设置了`maxFilesPerTrigger`，则将忽略此参数，因为有效且应处理的旧文件可能会被忽略。<br/>最大期限是根据最新文件的时间戳指定的，而不是当前系统的时间戳。（默认值：1 周）<br/><br/> `cleanSource`：选项，用于在处理后清理已完成的文件。可用选项包括"archive"、"delete"、"off"。<br/>如果未提供该选项，则默认值为"off"。 | 是 |
# MAGIC | **Socket源** | `host`：要连接到的主机，必须指定<br/><br/>`port`：要连接到的端口，必须指定 | 否 |
# MAGIC | **Rate源**   | `rowsPerSecond`（例如 100，默认值：1）：每秒应生成多少行。<br/><br/>`rampUpTime`（例如 5s，默认值：0s）：在生成速度变为`rowsPerSecond`之前需要多长时间上升。使用比秒更精细的粒度将被截断为整数秒。<br/><br/>`numPartitions`（例如 10，默认值：Spark的默认并行度）：生成的行的分区号。<br/><br/>源将尽力达到`rowsPerSecond`，但查询可能受资源限制，可以进行`numPartitions`调整以帮助达到所需的速度。| 是 |
# MAGIC | **Kafka源** | 请参阅[Kafka集成指南](https://spark.apache.org/docs/3.2.1/structured-streaming-kafka-integration.html)。 | 是 |

# COMMAND ----------

# MAGIC %md
# MAGIC 
# MAGIC 下面来看一些案例。

# COMMAND ----------

# MAGIC %md
# MAGIC 
# MAGIC #### 文件源

# COMMAND ----------

from pyspark.sql.types import StructType, StringType

schema = StructType().\
add("OrderNumber", StringType(), nullable=False).\
add("OrderDate", StringType(), nullable=False).\
add("ItemName", StringType(), nullable=False).\
add("Quantity", StringType(), nullable=False).\
add("ProductPrice", StringType(), nullable=False).\
add("TotalProducts", StringType(), nullable=False)

df = spark.readStream.option("sep", ",").schema(schema).format("csv").load("/mnt/databrickscontainer1")
# df = spark.readStream.option("sep", ",").schema(schema).csv("/mnt/databrickscontainer1/")

print(type(df))

# 想要查看、输出数据，必须使用writeStream.start()
# AnalysisException: Queries with streaming sources must be executed with writeStream.start();
# df.show()
# display(df)
print(df.isStreaming)

# 不带聚合操作直接显示数据
# format: console/memory
# query = df.writeStream.format("console").start()

# 带有聚合操作的，需要设置outputMode
df.createOrReplaceTempView("orders")
# query = spark.sql("select ItemName, count(*) from orders group by ItemName").writeStream.outputMode("complete").format("console").start()
display(spark.sql("select ItemName, count(*) from orders group by ItemName"))

# query.awaitTermination()

# COMMAND ----------

# MAGIC %sh
# MAGIC 
# MAGIC ping 104.208.119.0

# COMMAND ----------

# MAGIC %md
# MAGIC 
# MAGIC #### Socket源

# COMMAND ----------

import pyspark.sql.functions as F

# 获取行数据
lines = spark \
    .readStream \
    .format("socket") \
    .option("host", "104.208.119.0") \
    .option("port", 5555) \
    .load()

print(type(lines))

# 将行拆分成单词
words = lines.select(
   # 炸裂函数
   F.explode(
       F.split(lines.value, " ")
   ).alias("word")
)
print(type(words))

# 不带聚合操作直接显示数据
# query = words.writeStream.format("console").start()

# 带有聚合操作的，需要设置outputMode
# 统计单词的个数
wordCounts = words.groupBy("word").count()
print(type(wordCounts))
query = wordCounts.writeStream.outputMode("complete").format("console").start()

print(type(query))

# query.awaitTermination()

# COMMAND ----------

# MAGIC %md
# MAGIC 
# MAGIC #### Rate源
# MAGIC 
# MAGIC rowsPerSecond（例如 100，默认值：1）：每秒应生成多少行。
# MAGIC 
# MAGIC rampUpTime（例如 5s，默认值：0s）：在生成速度变为rowsPerSecond之前需要多长时间上升。使用比秒更精细的粒度将被截断为整数秒。
# MAGIC 
# MAGIC numPartitions（例如 10，默认值：Spark的默认并行度）：生成的行的分区号。
# MAGIC 
# MAGIC 源将尽力达到rowsPerSecond，但查询可能受资源限制，可以进行numPartitions调整以帮助达到所需的速度。

# COMMAND ----------

import pyspark.sql.functions as F

# 获取行数据
lines = spark \
    .readStream \
    .format("rate") \
    .option("rowsPerSecond", 10) \
    .option("rampUpTime", 1) \
    .option('includeTimestamp', 'true') \
    .load()

# 将行拆分成单词
words = lines.select(
   lines.value.alias("word"),
   lines.timestamp
).groupBy("timestamp").count().orderBy("timestamp", ascending=False)

query = words\
    .writeStream\
    .outputMode('complete')\
    .format('console')\
    .option("truncate", "false")\
    .start()

# COMMAND ----------

# MAGIC %md
# MAGIC 
# MAGIC #### Kafka源
# MAGIC 
# MAGIC [https://spark.apache.org/docs/3.2.1/structured-streaming-kafka-integration.html](https://spark.apache.org/docs/3.2.1/structured-streaming-kafka-integration.html)
# MAGIC 
# MAGIC #### Ubuntu20.04下Kafka安装与部署
# MAGIC 
# MAGIC ```
# MAGIC sudo apt-get update
# MAGIC ```
# MAGIC 
# MAGIC ##### 安装JDK
# MAGIC 
# MAGIC ```
# MAGIC sudo apt install openjdk-8-jdk
# MAGIC ```
# MAGIC 
# MAGIC ##### 下载Kafka
# MAGIC 
# MAGIC ```
# MAGIC wget https://downloads.apache.org/kafka/3.3.1/kafka_2.12-3.3.1.tgz
# MAGIC wget https://archive.apache.org/dist/kafka/3.1.0/kafka_2.12-3.1.0.tgz
# MAGIC ```
# MAGIC 
# MAGIC ##### 安装Kafka
# MAGIC 
# MAGIC ###### 解压
# MAGIC 
# MAGIC ```
# MAGIC mkdir apps
# MAGIC tar -xzf kafka_2.12-3.1.0.tgz -C apps/
# MAGIC ```
# MAGIC 
# MAGIC ###### 配置
# MAGIC 
# MAGIC ```
# MAGIC vi apps/kafka_2.12-3.1.0/config/zookeeper.properties
# MAGIC vi apps/kafka_2.12-3.1.0/config/server.properties
# MAGIC ```
# MAGIC 
# MAGIC ###### 启动
# MAGIC 
# MAGIC ```
# MAGIC cd apps/kafka_2.12-3.1.0
# MAGIC 
# MAGIC bin/zookeeper-server-start.sh -daemon config/zookeeper.properties
# MAGIC 
# MAGIC bin/kafka-server-start.sh -daemon config/server.properties
# MAGIC ```
# MAGIC 
# MAGIC ##### 使用Kafka
# MAGIC 
# MAGIC ###### 创建Topic
# MAGIC 
# MAGIC ```
# MAGIC bin/kafka-topics.sh --create --topic KafkaFirstTopic --partitions 1 --replication-factor 1 --bootstrap-server 10.0.0.4:9092
# MAGIC 
# MAGIC bin/kafka-topics.sh --list --bootstrap-server 10.0.0.4:9092
# MAGIC ```
# MAGIC 
# MAGIC ###### 发送消息
# MAGIC 
# MAGIC ```
# MAGIC bin/kafka-console-producer.sh --bootstrap-server 10.0.0.4:9092 --topic KafkaFirstTopic
# MAGIC ```
# MAGIC 
# MAGIC ###### 消费消息
# MAGIC 
# MAGIC ```
# MAGIC bin/kafka-console-consumer.sh --bootstrap-server 10.0.0.4:9092 --topic KafkaFirstTopic
# MAGIC ```

# COMMAND ----------

# MAGIC %sh
# MAGIC 
# MAGIC ping 104.208.119.0

# COMMAND ----------

# MAGIC %sh
# MAGIC 
# MAGIC cat /etc/hosts

# COMMAND ----------

# MAGIC %sh
# MAGIC 
# MAGIC echo "104.208.119.0 wux-labs-vm.internal.cloudapp.net" >> /etc/hosts
# MAGIC # echo "20.239.80.52 wux-labs-vm.internal.cloudapp.net" >> /etc/hosts

# COMMAND ----------

df = spark \
  .readStream \
  .format("kafka") \
  .option("kafka.bootstrap.servers", "104.208.119.0:9092") \
  .option("subscribe", "KafkaFirstTopic") \
  .load()

# 原始数据
df.writeStream.format("console").start()

# 对数据做转换
df.selectExpr("explode(split(CAST(value AS STRING),\" \"))").writeStream.format("console").start()

# 对数据做转换、统计
wordCount = df.selectExpr("explode(split(CAST(value AS STRING),\" \"))").groupBy("col").count()

wordCount.writeStream.outputMode("complete").format("console").start()

display(wordCount)

# query.awaitTermination()

# COMMAND ----------

# MAGIC %md
# MAGIC 
# MAGIC ### 流式 DataFrames/Datasets 的模式（Schema）推断和分区
# MAGIC 
# MAGIC 默认情况下，来自基于文件的源的结构化流式处理需要指定schema，而不是依赖Spark自动推断schema。此限制可确保将一致的schema用于流式处理查询，即使在失败的情况下也是如此。对于临时用例，可以通过将`spark.sql.streaming.schemaInference`设置为`true`。
# MAGIC 
# MAGIC 当存在名为/key=value/的子目录并且列表将自动递归到这些目录中时，分区发现确实会发生。如果这些列出现在用户提供的schema中，则Spark将根据正在读取的文件的路径填充这些列。组成分区方案的目录必须在查询开始时存在，并且必须保持静态。例如，当/data/year=2015/存在时，可以添加/data/year=2016/，但更改分区列（即通过创建目录/data/date=2016-04-17/）是无效的。

# COMMAND ----------

# MAGIC %md
# MAGIC 
# MAGIC ## 对流式DataFrames/Datasets的操作
# MAGIC 
# MAGIC 可以对流式DataFrames/Datasets应用各种操作，范围从非类型化的、SQL类的操作（例如select、where、groupBy）到类型化的RDD类操作（如map、filter、flatMap）。

# COMMAND ----------

# MAGIC %md
# MAGIC 
# MAGIC ### 基本操作 - 选择、投影、聚合
# MAGIC 
# MAGIC #### Selection, Projection, Aggregation
# MAGIC 
# MAGIC 流式处理支持对DataFrame/Dataset执行大多数常见操作。

# COMMAND ----------

df = spark \
  .readStream \
  .format("kafka") \
  .option("kafka.bootstrap.servers", "104.208.119.0:9092") \
  .option("subscribe", "KafkaFirstTopic") \
  .load()

# 原始数据
query = df.writeStream.format("console").start()

# 对数据做转换
df.selectExpr("explode(split(CAST(value AS STRING),\" \"))").writeStream.format("console").start()

# 对数据做转换、统计
df.selectExpr("explode(split(CAST(value AS STRING),\" \"))").groupBy("col").count().writeStream.outputMode("complete").format("console").start()


# COMMAND ----------

df = spark \
  .readStream \
  .format("kafka") \
  .option("kafka.bootstrap.servers", "104.208.119.0:9092") \
  .option("subscribe", "KafkaFirstTopic") \
  .load()

df.createOrReplaceTempView("topics")

spark.sql("select cast(value as string) as value from topics").createOrReplaceTempView("values")

spark.sql("select explode(split(value, ' ')) as word from values").createOrReplaceTempView("words")

spark.sql("select word,count(*) from words group by word having count(*) > 1").writeStream.outputMode("complete").format("console").start()

# COMMAND ----------

# MAGIC %md
# MAGIC 
# MAGIC ### 事件时间的窗口操作
# MAGIC 
# MAGIC 使用结构化流式处理，滑动事件时间窗口上的聚合非常简单，并且与分组聚合非常相似。在分组聚合中，为用户指定的分组列中的每个唯一值维护聚合值（例如计数）。对于基于窗口的聚合，将为行的事件时间所属的每个窗口维护聚合值。让我们通过插图来理解这一点。
# MAGIC 
# MAGIC 假设我们的入门示例被修改，流现在包含行以及生成行的时间。我们希望在 10 分钟内计算字数，每 5 分钟更新一次，而不是运行字数统计。也就是说，在 10 分钟窗口 12:00 - 12:10、12:05 - 12:15、12:10 - 12:20 等之间收到的字数。请注意，12:00 - 12:10 表示在 12:00 之后但 12:10 之前到达的数据。现在，考虑在 12:07 收到的一个词。此字应递增对应于两个窗口 12:00 - 12:10 和 12:05 - 12:15 的计数。因此，计数将按分组键（即单词）和窗口（可以从事件时间计算）进行索引。
# MAGIC 
# MAGIC ![](https://spark.apache.org/docs/3.2.1/img/structured-streaming-window.png)
# MAGIC 
# MAGIC 由于此窗口化类似于分组，因此在代码中，可以使用 groupBy() 和 window() 操作来表示窗口化聚合。

# COMMAND ----------

import pyspark.sql.functions as F

# 获取行数据
lines = spark \
    .readStream \
    .format("socket") \
    .option("host", "104.208.119.0") \
    .option("port", 5555) \
    .option('includeTimestamp', 'true') \
    .load()

# 将行拆分成单词
words = lines.select(
   # 炸裂函数
   F.explode(
       F.split(lines.value, " ")
   ).alias("word"),
   lines.timestamp
)

windowDuration = '10 seconds'
slideDuration = '5 seconds'

# 原始数据
# lines.writeStream \
#     .format('console') \
#     .option("truncate",False) \
#     .start()

# 划分窗口
# words.select(
#     words.word,
#     words.timestamp,
#     F.window(words.timestamp, windowDuration, slideDuration)
# ).writeStream \
#     .format('console') \
#     .option("truncate",False) \
#     .start()

# 窗口数据
words.groupBy(
    F.window(words.timestamp, windowDuration, slideDuration),
    words.word
).count().orderBy('window', ascending=False) \
    .writeStream\
    .outputMode('complete')\
    .format('console')\
    .option("truncate",False) \
    .start()

# COMMAND ----------

# MAGIC %md
# MAGIC 
# MAGIC #### 处理延迟数据和水印
# MAGIC 
# MAGIC 现在考虑如果其中一个事件迟到应用程序会发生什么情况。例如，应用程序可以在 12:11 接收在 12:04（即事件时间）生成的单词。应用程序应使用时间 12:04 而不是 12:11 来更新窗口 12:00 - 12:10 的旧计数。这在我们基于窗口的分组中自然发生 - 结构化流式处理可以长时间保持部分聚合的中间状态，以便后期数据可以正确更新旧窗口的聚合，如下图所示。
# MAGIC 
# MAGIC ![](https://spark.apache.org/docs/3.2.1/img/structured-streaming-late-data.png)
# MAGIC 
# MAGIC 然而，要连续几天运行此查询，系统必须绑定其累积的中间内存状态量。这意味着系统需要知道何时可以从内存状态中删除旧聚合，因为应用程序将不再接收该聚合的延迟数据。为了实现这一点，在Spark 2.1中，引入了watermarking（水印/水位线），它让引擎自动跟踪数据中的当前事件时间，并尝试相应地清除旧状态。可以通过指定事件时间列和数据在事件时间方面的延迟阈值来定义查询的水印。对于在时间T结束的特定窗口，引擎将保持状态并允许延迟数据更新状态，直到（引擎看到的最大事件时间-延迟阈值>T）。换言之，阈值内的延迟数据将被聚合，但超过阈值的数据将开始被丢弃。让我们用一个例子来理解这一点。我们可以使用下面所示的withWatermark()轻松地在前面的示例中定义水印。

# COMMAND ----------

# MAGIC %sh
# MAGIC 
# MAGIC date

# COMMAND ----------

import pyspark.sql.functions as F
import datetime

df = spark \
  .readStream \
  .format("kafka") \
  .option("kafka.bootstrap.servers", "104.208.119.0:9092") \
  .option("subscribe", "KafkaFirstTopic") \
  .load()

# 原始数据
df.writeStream.format("console").start()

# 消息和时间戳
df.select("value", "timestamp").writeStream.format("console").start()

# 拆分的消息和时间戳
df.select(F.split(df.value, ","), "timestamp").writeStream.format("console").start()

# 获取拆分后的单列数据和时间戳
split_df = F.split(df.value, ",")
df.select(split_df[0],split_df[1],split_df[2],"timestamp").writeStream.format("console").start()
df.select(split_df[0].alias("col1"),split_df[1].alias("col2"),F.to_timestamp(split_df[2]).alias("event_timestamp"),df["timestamp"].alias("kafka_receive_timestamp")).writeStream.format("console").start()


# COMMAND ----------

import pyspark.sql.functions as F

# 获取行数据
lines = spark \
    .readStream \
    .format("kafka") \
    .option("kafka.bootstrap.servers", "104.208.119.0:9092") \
    .option("subscribe", "KafkaFirstTopic") \
    .option('includeTimestamp', 'true') \
    .load()

# 原始数据
# lines.writeStream.format("console").option("truncate",False).start()

# 将行拆分成单词
words = lines.select(
   # 炸裂函数
   F.explode(
       F.split(lines.value, ",")
   ).alias("word"),
   F.to_timestamp(F.split(lines.value, ",")[2]).alias("event_timestamp"),
   lines.timestamp.alias("kafka_receive_timestamp")
).withWatermark("event_timestamp", "10 seconds")

words.writeStream.format("console").option("truncate",False).start()

windowDuration = '10 seconds'
slideDuration = '5 seconds'

# 完整模式
# words \
#     .groupBy(
#         F.window(words.event_timestamp, windowDuration, slideDuration),
#         words.word.alias("complete word")
#     ).count()\
#     .writeStream\
#     .outputMode('complete')\
#     .format('console')\
#     .option("truncate",False)\
#     .start()

# 更新模式
words \
    .groupBy(
        F.window(words.event_timestamp, windowDuration, slideDuration),
        words.word.alias("update word")
    ).count()\
    .writeStream\
    .outputMode('update')\
    .format('console')\
    .option("truncate",False)\
    .start()

# COMMAND ----------

# MAGIC %md
# MAGIC 
# MAGIC 发送一条数据，用来推进事件时间的移动
# MAGIC ```
# MAGIC data,from,2022-11-11 12:34:56
# MAGIC ```  
# MAGIC 
# MAGIC **对于更新模式：**
# MAGIC 
# MAGIC 事件时间移动以后，根据这个最大事件时间、延迟阈值，可以得到最大允许延迟数据的时间窗口的结束下限：34:56 - 10 = 34:46，也就是延迟数据的时间窗口的结束时间必须大于34:46，也就是结束时间是34:50，最小窗口是 34:40 - 34:50，则能够接收的最大延迟数据是 2022-11-11 12:34:40，因为它属于时间窗口 34:40 - 34:50
# MAGIC ```
# MAGIC < data,from,2022-11-11 12:34:40  数据会被丢弃
# MAGIC data,from,2022-11-11 12:34:39    数据会被丢弃
# MAGIC 
# MAGIC data,from,2022-11-11 12:34:40    数据会被接受
# MAGIC 
# MAGIC data,from,2022-11-11 12:34:41    数据会被接受
# MAGIC > data,from,2022-11-11 12:34:40  数据会被接受
# MAGIC ```
# MAGIC 
# MAGIC **对于完全模式：**
# MAGIC 
# MAGIC 数据都会被接受。

# COMMAND ----------

# MAGIC %md
# MAGIC 
# MAGIC 我们在"event_timestamp"列的值上定义查询的水印， 并将"10 minutes"（代码中我们使用的是"10 seconds"）定义为允许数据延迟的阈值。如果此查询在更新输出模式（OutputMode.Update）下运行， 引擎将不断更新结果表中窗口的计数，直到窗口早于水印，它比"event_timestamp"列中的当前事件时间滞后"10 seconds"。
# MAGIC 
# MAGIC ![](https://spark.apache.org/docs/3.2.1/img/structured-streaming-watermark-update-mode.png)
# MAGIC 
# MAGIC 如图所示，引擎跟踪的最大事件时间是蓝色虚线，在每个触发器开始时设置为（最大事件时间-'10 mins'）的水印是红线。例如，当引擎观察数据（12:14，dog）时，它将下一个触发器的水印设置为12:04。该水印允许引擎将中间状态再保持10分钟，以便对延迟的数据进行计数。例如，数据（12:09，cat）出现故障和延迟，并且在窗口12:00-12:10和12:05-12:15中出现。由于它仍在触发器中的水印12:04之前，引擎仍将中间计数保持为状态，并正确更新相关窗口的计数。然而，当水印被更新到12:11时，窗口（12:00-12:10）的中间状态被清除，并且所有后续数据（例如（12:04，donkey））被认为“太晚”，因此被忽略。请注意，在每次触发后，更新的计数（即紫色行）都会写入接收器作为触发输出，如更新模式所示。

# COMMAND ----------

# MAGIC %md
# MAGIC 
# MAGIC #### 时间窗口的类型
# MAGIC 
# MAGIC Spark 支持三种类型的时间窗口：滚动（固定）窗口、滑动窗口和会话窗口。
# MAGIC 
# MAGIC ![](https://spark.apache.org/docs/3.2.1/img/structured-streaming-time-window-types.jpg)
# MAGIC 
# MAGIC **滚动窗口**是一系列固定大小、不重叠且连续的时间间隔。输入只能绑定到单个窗口。
# MAGIC 
# MAGIC 从“固定大小”的角度来看，**滑动窗口**类似于**滚动窗口**，但如果滑动的持续时间小于窗口的持续时间，窗口可能会重叠，在这种情况下，输入可以绑定到多个窗口。
# MAGIC 
# MAGIC 滚动和滑动窗口使用窗口功能，这在上面的示例中已经描述过。
# MAGIC 
# MAGIC 与前两种类型相比，**会话窗口**具有不同的特性。会话窗口具有窗口长度的动态大小，具体取决于输入。会话窗口以输入开始，如果在间隙持续时间内接收到以下输入，则会自行扩展。对于静态间隙持续时间，当在接收到最新输入后的间隙持续时间内没有接收到输入时，会话窗口将关闭。
# MAGIC 
# MAGIC 会话窗口使用`session_window`函数。该函数的用法类似于`window`函数。

# COMMAND ----------

import pyspark.sql.functions as F

# 获取行数据
lines = spark \
    .readStream \
    .format("kafka") \
    .option("kafka.bootstrap.servers", "104.208.119.0:9092") \
    .option("subscribe", "KafkaFirstTopic") \
    .option('includeTimestamp', 'true') \
    .load()

# 原始数据
# lines.writeStream.format("console").option("truncate",False).start()

# 将行拆分成单词
words = lines.select(
   # 炸裂函数
   F.explode(
       F.split(lines.value, ",")
   ).alias("word"),
   F.to_timestamp(F.split(lines.value, ",")[2]).alias("event_timestamp"),
   lines.timestamp.alias("kafka_receive_timestamp")
).withWatermark("event_timestamp", "10 seconds")

words.writeStream.format("console").option("truncate",False).start()

windowDuration = '10 seconds'

# 完整模式
words \
    .groupBy(
        F.session_window(words.event_timestamp, windowDuration),
        words.word
    ).count()\
    .writeStream\
    .outputMode('complete')\
    .format('console')\
    .option("truncate",False)\
    .start()

# COMMAND ----------

# MAGIC %md
# MAGIC 
# MAGIC 请注意，在流式查询中使用会话窗口时有一些限制，如下所示：
# MAGIC 
# MAGIC * 不支持“更新模式”作为输出模式。
# MAGIC > Update output mode not supported for session window on streaming DataFrames/DataSets
# MAGIC * 分组键中除了session_window之外，至少应该有一列。
# MAGIC > Global aggregation with session window in streaming query is not supported
# MAGIC * 对于批处理查询，支持全局窗口（分组键中只有session_window）。

# COMMAND ----------

import pyspark.sql.functions as F

# 获取行数据
lines = spark \
    .readStream \
    .format("kafka") \
    .option("kafka.bootstrap.servers", "104.208.119.0:9092") \
    .option("subscribe", "KafkaFirstTopic") \
    .option('includeTimestamp', 'true') \
    .load()

# 原始数据
# lines.writeStream.format("console").option("truncate",False).start()

# 将行拆分成单词
words = lines.select(
   # 炸裂函数
   F.explode(
       F.split(lines.value, ",")
   ).alias("word"),
   F.to_timestamp(F.split(lines.value, ",")[2]).alias("event_timestamp"),
   lines.timestamp.alias("kafka_receive_timestamp")
).withWatermark("event_timestamp", "10 seconds")

words.writeStream.format("console").option("truncate",False).start()

windowDuration = '10 seconds'

# 完整模式
# Update output mode not supported for session window on streaming DataFrames/DataSets;
# Global aggregation with session window in streaming query is not supported.
words \
    .groupBy(
        F.session_window(words.event_timestamp, windowDuration),
        words.word
    ).count()\
    .writeStream\
    .outputMode('complete')\
    .format('console')\
    .option("truncate",False)\
    .start()

# COMMAND ----------

# MAGIC %md
# MAGIC 
# MAGIC #### 水印清除聚合状态的条件
# MAGIC 
# MAGIC 需要注意的是，水印必须满足以下条件才能清除查询中的聚合状态（从Spark 2.1.1开始，可能会在将来发生更改）。
# MAGIC 
# MAGIC * 输出模式必须为“Append”或“Update”。Complete模式要求保留所有聚合数据，因此不能使用水印来删除中间状态。
# MAGIC * 聚合必须具有事件时间列或事件时间列上的窗口。
# MAGIC * 必须在与聚合中使用的时间戳列相同的列上调用withWatermark。例如，df.withWatermark("event_timestamp", "10 seconds").groupBy("kafka_receive_timestamp").count()在Append输出模式下无效，因为水印是在与聚合列不同的列上定义的。
# MAGIC * 必须在聚合之前调用withWatermark才能使用水印详细信息。例如，df.groupBy("event_timestamp").count().withWatermark("event_timestamp", "10 seconds")在Append输出模式下无效。

# COMMAND ----------

# MAGIC %md
# MAGIC 
# MAGIC #### 水印聚合的语义保证
# MAGIC 
# MAGIC * 水印延迟（使用水印设置）为“2 hours”可确保引擎不会丢弃任何延迟小于2小时的数据。换言之，任何数据（就事件时间而言）比最新处理的数据晚不到2个小时的，都保证会被聚合。
# MAGIC * 然而，这种保证只在一个方向上是严格的。延迟超过2小时的数据不保证被丢弃；它可能会聚集，也可能不会聚集。数据越延迟，引擎处理数据的可能性就越小。

# COMMAND ----------

# MAGIC %md
# MAGIC 
# MAGIC ### 连接（Join）操作
# MAGIC 
# MAGIC Structured Streaming支持将流式Dataset/DataFrame与静态Dataset/DataFrame以及另一个流式Dataset/DataFrame连接起来。流连接的结果是增量生成的，类似于上一节中的流聚合结果。我们将探讨在上述情况下支持什么类型的连接（即inner、outer、semi等）。请注意，在所有支持的连接类型中，与流式Dataset/DataFrame的连接结果将完全相同，就像与流中包含相同数据的静态Dataset/DataFrame的连接一样。

# COMMAND ----------

# MAGIC %md
# MAGIC 
# MAGIC #### 流-静态数据连接
# MAGIC 
# MAGIC 自Spark 2.0引入以来，Structured Streaming支持流和静态Dataset/DataFrame之间的连接（内部连接和某种类型的外部连接）。
# MAGIC 
# MAGIC > 流-静态数据连接不是有状态的，因此不需要状态管理。

# COMMAND ----------

staticDf = spark.read.format("orc").load("/mnt/databrickscontainer1/restaurant-1-orders.orc").where("OrderNumber = '16118'")

lines = spark \
    .readStream \
    .format("rate") \
    .option("rowsPerSecond", 1) \
    .option("rampUpTime", 10) \
    .option('includeTimestamp', 'true') \
    .load()

words = lines.select(
   lines.value.alias("Quantity"),
   lines.timestamp
)

# 没有指定关联条件，笛卡尔积
# display(words.join(staticDf))
# words.join(staticDf).writeStream.outputMode('complete').format('console').option("truncate",False).start()

# 指定关联条件，没有指定关联方式，默认 Inner Join
# display(words.join(staticDf, "Quantity"))
# words.join(staticDf, "Quantity").writeStream.outputMode('complete').format('console').option("truncate",False).start()

# 指定关联条件，指定关联方式
display(words.join(staticDf, "Quantity", "left"))
# words.join(staticDf, "Quantity", "left").writeStream.outputMode('complete').format('console').option("truncate",False).start()

# COMMAND ----------

# MAGIC %md
# MAGIC 
# MAGIC #### 流-流连接
# MAGIC 在Spark 2.3中，增加了对流-流连接的支持，即可以连接两个流式Dataset/DataFrame。在两个数据流之间生成连接结果的挑战在于，在任何时间点，连接双方的数据集视图都不完整，在输入之间找到匹配项要困难得多。从一个输入流接收的任何行都可以匹配来自其他输入流的任何未来尚未接收的行。因此，对于两个输入流，我们将过去的输入缓冲为流状态，以便我们可以将每个未来的输入与过去的输入连接并相应地生成连接结果。此外，与流聚合类似， 我们会自动处理延迟、无序的数据，并可以使用水印限制状态。

# COMMAND ----------

# MAGIC %md
# MAGIC 
# MAGIC ##### 带可选水印的内连接
# MAGIC 
# MAGIC 支持任何类型的列上的内部连接以及任何类型的连接条件。但是，随着流的运行，流状态的大小将无限增长，因为必须保存**所有**过去的输入，因为任何新输入都可以与过去的任何输入匹配。 为了避免无界状态，必须定义额外的连接条件，以便无限期的旧输入不能与未来的输入匹配，因此可以从状态中清除。换句话说，您必须在连接中执行以下附加步骤。
# MAGIC 
# MAGIC 1. 在两个输入上定义水印延迟，以便引擎知道输入的延迟程度（类似于流式聚合）
# MAGIC 2. 定义两个输入之间的事件时间约束，以便引擎可以确定何时不再需要一个输入的旧行（即不满足时间约束）来匹配另一个输入。该约束可以通过以下两种方式之一定义。
# MAGIC     1. 时间范围连接条件（例如），...JOIN... ON leftTime BETWEEN rightTime AND rightTime + INTERVAL 1 HOUR
# MAGIC     2. 在事件时间窗口上进行连接（例如）。...JOIN... ON leftTimeWindow = rightTimeWindow
# MAGIC 
# MAGIC 我们来看一个例子。
# MAGIC 
# MAGIC 假如我们需要统计餐饮店在发布新菜品后的一个小时内的用户下单信息，那么我们会有两个流：一个是餐饮店发布新菜品的流，一个是用户下单的流。两个流的连接我们就可以得到菜品的下单信息，若要允许在此流-流连接中进行状态清理，必须指定水印延迟和时间限制。
# MAGIC 
# MAGIC * 水印延迟：菜品的发布和用户的下单都有可能有延迟，分别最多2小时和3小时。
# MAGIC * 事件时间范围条件：我们只统计新菜品发布后的一个小时内的下单信息。
# MAGIC 
# MAGIC > * 水印在流-流的内连接中是可选的
# MAGIC > * “2小时”水印延迟保证引擎永远不会丢弃任何小于延迟2小时。但是延迟超过2小时的数据可能会也可能不会被处理。

# COMMAND ----------

import pyspark.sql.functions as F

# 菜品发布流
# 104.208.119.0
items = spark \
    .readStream \
    .format("socket") \
    .option("host", "104.208.119.0") \
    .option("port", 5555) \
    .load()

# 用户下单流
orders = spark \
    .readStream \
    .format("socket") \
    .option("host", "104.208.119.0") \
    .option("port", 6666) \
    .load()

itemsWithWatermark = items.select(
   F.split(items.value, ",")[0].alias("item_no"),
   F.split(items.value, ",")[1].alias("item_name"),
   F.split(items.value, ",")[2].alias("item_desc"),
   F.to_timestamp(F.split(items.value, ",")[3]).alias("item_timestamp")
)#.withWatermark("item_timestamp", "2 hours")

ordersWithWatermark = orders.select(
   F.split(orders.value, ",")[0].alias("user_name"),
   F.split(orders.value, ",")[1].alias("order_item_no"),
   F.to_timestamp(F.split(orders.value, ",")[2]).alias("order_timestamp")
)#.withWatermark("order_timestamp", "3 hours")

joins = itemsWithWatermark.join(ordersWithWatermark, F.expr("order_item_no = item_no and order_timestamp >= item_timestamp and order_timestamp <= item_timestamp + interval 1 hour"))

display(joins)
# joins.writeStream.format('console').option("truncate",False).start()

# COMMAND ----------

# MAGIC %md
# MAGIC 
# MAGIC 菜品流水
# MAGIC ```
# MAGIC 1,Garlic Naan,蒜蓉薄饼,2022-11-18 12:23:34
# MAGIC 2,Mushroom Rice,蘑菇饭,2022-11-18 12:34:56
# MAGIC ```
# MAGIC 
# MAGIC 下单流水
# MAGIC ```
# MAGIC user1,1,2022-11-18 12:23:34
# MAGIC user2,1,2022-11-18 13:23:33
# MAGIC user3,1,2022-11-18 13:23:35
# MAGIC user1,2,2022-11-18 12:23:45
# MAGIC user2,2,2022-11-18 13:34:56
# MAGIC user4,2,2022-11-18 13:34:57
# MAGIC ```

# COMMAND ----------

# MAGIC %md
# MAGIC 
# MAGIC ##### 带水印的外连接
# MAGIC 
# MAGIC 虽然水印+事件时间约束对于内连接是可选的，但是对于外连接，必须指定它们。这是因为为了在外连接中生成NULL结果，引擎必须知道输入行将来什么时候不匹配。因此，必须指定水印+事件时间约束以生成正确的结果。
# MAGIC 
# MAGIC > 外连接在水印延迟以及是否删除数据方面与内连接具有相同的语义保证。

# COMMAND ----------

import pyspark.sql.functions as F

# 菜品发布流
# 104.208.119.0
items = spark \
    .readStream \
    .format("socket") \
    .option("host", "104.208.119.0") \
    .option("port", 5555) \
    .load()
# items = spark \
#   .readStream \
#   .format("kafka") \
#   .option("kafka.bootstrap.servers", "104.208.119.0:9092") \
#   .option("subscribe", "ItemsTopic") \
#   .load()

# 用户下单流
orders = spark \
    .readStream \
    .format("socket") \
    .option("host", "104.208.119.0") \
    .option("port", 6666) \
    .load()
# orders = spark \
#   .readStream \
#   .format("kafka") \
#   .option("kafka.bootstrap.servers", "104.208.119.0:9092") \
#   .option("subscribe", "OrdersTopic") \
#   .load()


itemsWithWatermark = items.select(
   F.split(items.value, ",")[0].alias("item_no"),
   F.split(items.value, ",")[1].alias("item_name"),
   F.split(items.value, ",")[2].alias("item_desc"),
   F.to_timestamp(F.split(items.value, ",")[3]).alias("item_timestamp")
)#.withWatermark("item_timestamp", "2 hours")

ordersWithWatermark = orders.select(
   F.split(orders.value, ",")[0].alias("user_name"),
   F.split(orders.value, ",")[1].alias("order_item_no"),
   F.to_timestamp(F.split(orders.value, ",")[2]).alias("order_timestamp")
).withWatermark("order_timestamp", "3 hours")
# Stream-stream LeftOuter join between two streaming DataFrame/Datasets is not supported without a watermark in the join keys, or a watermark on the nullable side and an appropriate range condition;

# can be "inner", "leftOuter", "rightOuter", "fullOuter", "leftSemi"
joins = itemsWithWatermark.join(ordersWithWatermark, F.expr("order_item_no = item_no and order_timestamp >= item_timestamp and order_timestamp <= item_timestamp + interval 1 hour"),"leftOuter")

display(joins)
# joins.writeStream.outputMode("append").format('console').option("truncate",False).start()

# COMMAND ----------

# MAGIC %md
# MAGIC 
# MAGIC **注意** 
# MAGIC 
# MAGIC 关于如何生成外部结果，有几个重要的特征需要注意。
# MAGIC 
# MAGIC * 外部NULL结果将以取决于指定水印延迟和时间范围条件的延迟生成。这是因为引擎必须等待这么长时间，以确保没有匹配的数据，并且将来不会再有匹配的数据。简而言之，就是要等待过了延迟阈值才会生成NULL结果。
# MAGIC * 在微批处理引擎中的当前实现中，水印在微批结束时被提前，下一个微批处理使用更新的水印来清理状态并输出外部结果。由于我们仅在有新数据要处理时才触发微批处理，因此如果流中没有接收到新数据，则外部结果的生成可能会延迟。简而言之，如果被连接的两个输入流中的任何一个在一段时间内没有接收到数据，则外部（两种情况，左侧或右侧）输出可能会延迟。

# COMMAND ----------

# MAGIC %md
# MAGIC 
# MAGIC ##### 支持流式处理查询中连接的矩阵
# MAGIC 
# MAGIC | 左输入 | 右输入 | 连接类型 |  |
# MAGIC | ----- | ----- | ----- | ----- |
# MAGIC | 静态的 | 静态的 | 所有类型 | 支持，因为它不在流数据上，即使它可以存在于流式处理查询中</td>
# MAGIC | 流 | 静态的 | 内 | 支持，无状态 |
# MAGIC | 流 | 静态的 | 左外 | 支持，无状态 |
# MAGIC | 流 | 静态的 | 右外 | 不支持 |
# MAGIC | 流 | 静态的 | 全外 | 不支持 |
# MAGIC | 静态的 | 流 | 内 | 支持，无状态 | 
# MAGIC | 静态的 | 流 | 左外 | 不支持 |
# MAGIC | 静态的 | 流 | 右外 | 支持，无状态 |
# MAGIC | 静态的 | 流 | 全外 | 不支持 |
# MAGIC | 流 | 流 | 内 | 支持，可选双面指定水印+ 状态清理的时间限制 |
# MAGIC | 流 | 流 | 左外 | 有条件支持，必须在右侧指定水印 + 时间限制才能正确结果，（可选）为所有状态清理指定左侧的水印 |
# MAGIC | 流 | 流 | 右外 | 有条件支持，必须在左侧指定水印 + 时间限制才能正确结果，可以选择在右侧为所有状态清理指定水印 |
# MAGIC | 流 | 流 | 全外 | 有条件支持，必须在一侧指定水印 + 时间限制才能正确结果，可以选择在另一侧为所有状态清理指定水印 |
# MAGIC 
# MAGIC 有关支持的连接的其他详细信息：
# MAGIC 
# MAGIC 连接可以级联，也就是说，你可以这样做：df1.join(df2, ...).join(df3, ...).join(df4, ....)
# MAGIC 
# MAGIC * 从Spark 2.4开始，仅当查询处于追加输出模式时，才能使用连接。尚不支持其他输出模式。
# MAGIC * 从Spark 2.4开始，在连接之前不能使用其他非类似map的操作。以下是一些示例 什么不能使用。
# MAGIC   * 不能在连接之前使用流式聚合。
# MAGIC   * 在连接之前，不能在更新模式下使用mapGroupsWithState和flatMapGroupsWithState。

# COMMAND ----------

# MAGIC %md
# MAGIC 
# MAGIC ### 流式重复数据删除
# MAGIC 
# MAGIC 您可以使用事件中的唯一标识符对数据流中的记录进行重复数据消除。这与使用唯一标识符列的静态重复数据消除完全相同。查询将存储以前记录中所需的数据量，以便可以过滤重复记录。与聚合类似，您可以使用带或不带水印的重复数据消除。
# MAGIC 
# MAGIC * 带水印 - 如果重复记录到达的时间有上限，则可以在事件时间列上定义水印，并使用唯一标识符列和事件时间列进行重复数据消除。查询将使用水印从过去的记录中删除旧的状态数据，这些记录预计不会再获得任何重复数据。这限制了查询必须保持的状态量。
# MAGIC * 没有水印 - 由于重复记录可能到达的时间没有界限，因此查询将所有过去记录中的数据存储为状态。

# COMMAND ----------

# MAGIC %md
# MAGIC 
# MAGIC 假如我们要统计餐饮店下单用户的数量，那么我们就需要按用户ID来进行统计，而不是按照用户下了多少个订单来统计，此时我们应该使用user_name来进行去重。
# MAGIC 
# MAGIC 下单流水
# MAGIC ```
# MAGIC user1,1,2022-11-18 12:23:34
# MAGIC user2,1,2022-11-18 13:23:33
# MAGIC user3,1,2022-11-18 13:23:35
# MAGIC user1,2,2022-11-18 14:23:45
# MAGIC user2,2,2022-11-18 15:34:56
# MAGIC user4,2,2022-11-18 15:34:57
# MAGIC ```

# COMMAND ----------

import pyspark.sql.functions as F

# 用户下单流
# 104.208.119.0
orders = spark \
  .readStream \
  .format("kafka") \
  .option("kafka.bootstrap.servers", "20.239.80.52:9092") \
  .option("subscribe", "OrdersTopic") \
  .load()

# not isinstance(subset, Iterable) or isinstance(subset, str)
# TypeError("Parameter 'subset' must be a list of columns")
orderDropDuplicates = orders.select(
   F.split(orders.value, ",")[0].alias("user_name"),
   F.split(orders.value, ",")[1].alias("order_item_no"),
   F.to_timestamp(F.split(orders.value, ",")[2]).alias("order_timestamp")
).dropDuplicates(["user_name"])

orders.writeStream.format("console").option("truncate","false").start()
orderDropDuplicates.writeStream.format("console").option("truncate","false").start()

display(orderDropDuplicates)

# COMMAND ----------

# MAGIC %md
# MAGIC 
# MAGIC 下面看看带有水印设置的重复数据删除效果。
# MAGIC 
# MAGIC 下单流水
# MAGIC ```
# MAGIC user1,1,2022-11-18 12:23:34
# MAGIC user2,1,2022-11-18 13:23:33
# MAGIC user3,1,2022-11-18 13:23:35
# MAGIC user1,2,2022-11-18 14:23:45
# MAGIC user2,2,2022-11-18 15:34:56
# MAGIC user4,2,2022-11-18 15:34:57
# MAGIC ```

# COMMAND ----------

import pyspark.sql.functions as F

# 用户下单流
# 104.208.119.0
orders = spark \
  .readStream \
  .format("kafka") \
  .option("kafka.bootstrap.servers", "20.239.80.52:9092") \
  .option("subscribe", "OrdersTopic") \
  .load()

# not isinstance(subset, Iterable) or isinstance(subset, str)
# TypeError("Parameter 'subset' must be a list of columns")
orderDropDuplicates = orders.select(
   F.split(orders.value, ",")[0].alias("user_name"),
   F.split(orders.value, ",")[1].alias("order_item_no"),
   F.to_timestamp(F.split(orders.value, ",")[2]).alias("order_timestamp")
).withWatermark("order_timestamp", "1 hours").dropDuplicates(["user_name"])

orders.writeStream.format("console").option("truncate","false").start()
orderDropDuplicates.writeStream.format("console").option("truncate","false").start()

display(orderDropDuplicates)

# COMMAND ----------

# MAGIC %md
# MAGIC 
# MAGIC ### 处理多个水印的策略
# MAGIC 
# MAGIC 流式查询可以有多个输入流，这些输入流被联合或连接在一起。每个输入流可以具有不同的延迟数据阈值，比如前面的例子中，菜品发布的流和用户下单的流，分别具有2小时和3小时的延迟阈值，这对于有状态操作是需要容忍的。可以在每个输入流上使用`withWatermarks("eventTime", 延迟阈值)`指定这些阈值。
# MAGIC 
# MAGIC ```
# MAGIC itemsWithWatermark = items.withWatermark("item_timestamp", "2 hours")
# MAGIC 
# MAGIC ordersWithWatermark = orders.withWatermark("order_timestamp", "3 hours")
# MAGIC 
# MAGIC joins = itemsWithWatermark.join(ordersWithWatermark, joinCondition)
# MAGIC ```
# MAGIC 
# MAGIC 在执行查询时，结构化流单独跟踪每个输入流中看到的最大事件时间，根据相应的延迟计算水印，并选择一个全局水印用于有状态操作。
# MAGIC 
# MAGIC 默认情况下，选择最小值(多个输入流的水印的最小水印值)作为全局水印，因为它可以确保如果其中一个流落后于其他流（例如，其中一个由于上游故障而停止接收数据），则不会因太晚而意外丢弃数据。换句话说，**全局水印将以最慢流的速度安全地移动，查询输出将相应地延迟。**
# MAGIC 
# MAGIC 然而，在某些情况下，可能希望获得更快的结果，即使这意味着从最慢的流中删除数据。从Spark 2.4开始，可以通过设置SQL配置来设置多重水印策略，将`spark.sql.streaming.multipleWatermarkPolicy`设置为`max`（默认值为`min`）以选择最大值作为全局水印。这使得全局水印以最快的速度移动。然而，作为一个副作用，来自较慢流的数据将被大量丢弃。因此，请明智地使用此配置。

# COMMAND ----------

# MAGIC %md
# MAGIC 
# MAGIC ### 不支持的操作
# MAGIC 
# MAGIC 流式DataFrame/Dataset不支持一些DataFrame/Dataset操作。其中一些如下：
# MAGIC * 流数据集尚不支持多个流聚合（即流DF上的聚合链）。
# MAGIC * 流式数据集不支持限制和获取前N行。
# MAGIC * 不支持对流式处理数据集执行不同的操作。
# MAGIC * 在流数据集上聚合后，不支持重复数据删除操作。
# MAGIC * 只有在聚合后且在完全输出（"complete"）模式下，流式数据集才支持排序操作。
# MAGIC * 流数据集上不支持几种类型的外部联接。
# MAGIC 
# MAGIC 此外，还有一些数据集方法在流数据集上不起作用。这些操作将立即运行查询并返回结果，这在流数据集上没有意义。相反，这些功能可以通过显式启动流式查询来实现。
# MAGIC * count() - 无法从流数据集中返回单个计数。而是使用ds.groupBy().count()，返回包含运行计数的流数据集。
# MAGIC * foreach() - 改用writeStream.foreach(...)。
# MAGIC * show() - 改用writeStream.format("console")。
# MAGIC 
# MAGIC 如果尝试任何这些操作，将看到AnalysisException，如“operation XYZ is not supported with streaming DataFrames/Datasets”。虽然Spark的未来版本可能会支持其中的一些，但还有一些基本上很难在流数据上高效实现。例如，不支持对输入流进行排序，因为它需要跟踪流中接收的所有数据。因此，这从根本上来说很难有效执行。

# COMMAND ----------

# MAGIC %md
# MAGIC 
# MAGIC ## 启动流式处理查询
# MAGIC 
# MAGIC 一旦定义了最终结果DataFrame/Dataset，剩下的就是开始流计算。为此，必须使用通过Dataset.writeStream()返回的DataStreamWriter。必须在此接口中指定以下一项或多项。
# MAGIC 
# MAGIC * 输出模式：指定写入输出接收器的内容。
# MAGIC * 输出接收器的详细信息：format、location等。
# MAGIC * 查询名称：可以选择指定查询的唯一名称以进行标识。
# MAGIC * 触发间隔：可以选择指定触发间隔。如果未指定，系统将在上一处理完成后立即检查新数据的可用性。如果由于先前的处理尚未完成而错过了触发时间，则系统将立即触发处理。
# MAGIC * 检查点位置：对于可以保证端到端容错的某些输出接收器，请指定系统将写入所有检查点信息的位置。这应该是HDFS兼容容错文件系统中的一个目录。

# COMMAND ----------

# MAGIC %md
# MAGIC 
# MAGIC ### 输出模式（Output Modes）
# MAGIC 
# MAGIC 有几种类型的输出模式。
# MAGIC 
# MAGIC * 追加模式（Append，默认） - 这是默认模式，其中只有自上次触发以来添加到结果表的新行将被输出到接收器。只有那些添加到结果表中的行永远不会更改的查询才支持这一点。因此，这种模式保证每一行只输出一次（假设容错接收器）。例如，只有select、where、map、flatMap、filter、join等的查询将支持追加模式。
# MAGIC * 完整模式（Complete） - 每次触发后，整个结果表都将输出到接收器。聚合查询支持此功能。
# MAGIC * 更新模式（Update） - （从Spark 2.1.1开始可用）只有结果表中自上次触发后更新的行将被输出到接收器。更多信息将在未来版本中添加。
# MAGIC 
# MAGIC 不同类型的流式处理查询支持不同的输出模式。 下面是兼容性矩阵。
# MAGIC 
# MAGIC | 查询类型 | 支持的输出模式| 说明 |
# MAGIC | ----- | ----- | ----- |
# MAGIC | 使用聚合的查询(带水印的事件时间聚合) | Append<br/>Update<br/>Complete | 追加模式使用水印删除旧的聚合状态。<br/>更新模式使用水印删除旧的聚合状态。<br/>完整模式不会删除旧的聚合状态，因为根据定义，此模式 保留结果表中的所有数据。 |
# MAGIC | 使用聚合的查询（其他聚合） | Complete<br/>Update | 由于没有定义水印（仅在其他类别中定义），不会删除旧的聚合状态。 <br/>不支持追加模式，因为聚合可以更新，从而违反此模式。 |
# MAGIC | 带有`mapGroupsWithState`的查询 | Update | 带有`mapGroupsWithState`的查询不允许聚合。 |
# MAGIC | 带有`flatMapGroupsWithState`的查询（追加操作模式） | Append | `flatMapGroupsWithState`之后允许聚合。 |
# MAGIC | 带有`flatMapGroupsWithState`的查询（更新操作模式） | Update | `flatMapGroupsWithState`查询中不允许聚合。 |
# MAGIC | 带有`joins`的查询 | Append | 尚不支持更新和完整模式。 |
# MAGIC | 其他查询 | Append<br/>Update | 不支持完整模式，因为将所有未聚合的数据保留在结果表中是不可行的。 |

# COMMAND ----------

# MAGIC %md
# MAGIC 
# MAGIC ### 输出接收器（Output Sinks）
# MAGIC 
# MAGIC 有几种类型的内置输出接收器。
# MAGIC 
# MAGIC * **File sink** - 将输出存储到目录。
# MAGIC ```
# MAGIC writeStream
# MAGIC     .format("parquet")        // can be "orc", "json", "csv", etc.
# MAGIC     .option("path", "path/to/destination/dir")
# MAGIC     .start()
# MAGIC ```
# MAGIC * **Kafka sink** - 将输出存储在 Kafka 中的一个或多个主题。
# MAGIC ```
# MAGIC writeStream
# MAGIC     .format("kafka")
# MAGIC     .option("kafka.bootstrap.servers", "host1:port1,host2:port2")
# MAGIC     .option("topic", "updates")
# MAGIC     .start()
# MAGIC ```
# MAGIC * **Foreach sink** - 对输出中的记录运行任意计算。
# MAGIC ```
# MAGIC writeStream
# MAGIC     .foreach(...)
# MAGIC     .start()
# MAGIC ```
# MAGIC * **Console sink (for debugging)** - 每次有触发器时将输出打印到控制台/标准输出。支持追加和完整输出模式。这应该用于在低数据量上进行调试，因为每次触发后都会收集整个输出并将其存储在驱动程序的内存中。
# MAGIC ```
# MAGIC writeStream
# MAGIC     .format("console")
# MAGIC     .start()
# MAGIC ```
# MAGIC * **Memory sink (for debugging)** - 输出作为内存中表存储在内存中。 支持追加和完整输出模式。这应该用于调试目的 在低数据量下，因为整个输出被收集并存储在驱动程序的内存中。 因此，请谨慎使用它。
# MAGIC ```
# MAGIC writeStream
# MAGIC     .format("memory")
# MAGIC     .queryName("tableName")
# MAGIC     .start()
# MAGIC ```
# MAGIC 
# MAGIC 某些接收器不具有容错能力，因为它们不能保证输出的持久性，并且仅用于调试目的。
# MAGIC 
# MAGIC | Sink                  | Supported Output Modes   | Options                                                      | Fault-tolerant                                               | Notes                                                        |
# MAGIC | :-------------------- | :----------------------- | :----------------------------------------------------------- | :----------------------------------------------------------- | :----------------------------------------------------------- |
# MAGIC | **File Sink**         | Append                   | `path`: path to the output directory, must be specified. `retention`: time to live (TTL) for output files. Output files which batches were committed older than TTL will be eventually excluded in metadata log. This means reader queries which read the sink's output directory may not process them. You can provide the value as string format of the time. (like "12h", "7d", etc.) By default it's disabled.  For file-format-specific options, see the related methods in DataFrameWriter ([Scala](https://spark.apache.org/docs/3.2.1/api/scala/org/apache/spark/sql/DataFrameWriter.html)/[Java](https://spark.apache.org/docs/3.2.1/api/java/org/apache/spark/sql/DataFrameWriter.html)/[Python](https://spark.apache.org/docs/3.2.1/api/python/reference/api/pyspark.sql.streaming.DataStreamWriter.html#pyspark.sql.streaming.DataStreamWriter)/[R](https://spark.apache.org/docs/3.2.1/api/R/write.stream.html)). E.g. for "parquet" format options see `DataFrameWriter.parquet()` | Yes (exactly-once)                                           | Supports writes to partitioned tables. Partitioning by time may be useful. |
# MAGIC | **Kafka Sink**        | Append, Update, Complete | See the [Kafka Integration Guide](https://spark.apache.org/docs/3.2.1/structured-streaming-kafka-integration.html) | Yes (at-least-once)                                          | More details in the [Kafka Integration Guide](https://spark.apache.org/docs/3.2.1/structured-streaming-kafka-integration.html) |
# MAGIC | **Foreach Sink**      | Append, Update, Complete | None                                                         | Yes (at-least-once)                                          | More details in the [next section](https://spark.apache.org/docs/3.2.1/structured-streaming-programming-guide.html#using-foreach-and-foreachbatch) |
# MAGIC | **ForeachBatch Sink** | Append, Update, Complete | None                                                         | Depends on the implementation                                | More details in the [next section](https://spark.apache.org/docs/3.2.1/structured-streaming-programming-guide.html#using-foreach-and-foreachbatch) |
# MAGIC | **Console Sink**      | Append, Update, Complete | `numRows`: Number of rows to print every trigger (default: 20) `truncate`: Whether to truncate the output if too long (default: true) | No                                                           |                                                              |
# MAGIC | **Memory Sink**       | Append, Complete         | None                                                         | No. But in Complete Mode, restarted query will recreate the full table. | Table name is the query name.                                |
# MAGIC 
# MAGIC > 注意，必须调用`start()`来实际开始执行查询。这将返回StreamingQuery对象，该对象是持续运行的执行的句柄，可以使用此对象来管理查询。

# COMMAND ----------

# bin/kafka-console-producer.sh --bootstrap-server 10.0.0.4:9092 --topic KafkaFirstTopic
df = spark \
  .readStream \
  .format("kafka") \
  .option("kafka.bootstrap.servers", "20.239.80.52:9092") \
  .option("subscribe", "KafkaFirstTopic") \
  .load()

# Data source orc does not support Complete output mode
df.selectExpr("key","cast(value as string)","topic","partition","offset","timestamp").writeStream.format("csv").option("checkpointLocation", "/mnt/databrickscontainer1/checkpoint_file").option("path", "/mnt/databrickscontainer1/filesink").start()

# 对数据做转换、统计
wordCount = df.selectExpr("explode(split(CAST(value AS STRING),\" \"))").groupBy("col").count()

wordCount.writeStream.outputMode("complete").format("console").start()

# org.apache.spark.sql.AnalysisException: Required attribute 'value' not found
wordCount.selectExpr("col as value","count").writeStream.outputMode("complete").format("kafka").option("kafka.bootstrap.servers", "20.239.80.52:9092").option("topic", "KafkaSinkTopic").option("checkpointLocation", "/mnt/databrickscontainer1/checkpoint_kafka").start()
# bin/kafka-console-consumer.sh --bootstrap-server 10.0.0.4:9092 --topic KafkaSinkTopic

display(wordCount)

# COMMAND ----------

# MAGIC %md
# MAGIC 
# MAGIC #### 使用Foreach和ForeachBatch
# MAGIC 
# MAGIC `foreach`和`foreachBatch`操作允许对流查询的输出应用任意操作和编写逻辑。它们的用例略有不同，`foreach`允许在每一行上定制写逻辑，`foreachBatch`允许在每个微批处理的输出上进行任意操作和定制逻辑。

# COMMAND ----------

# MAGIC %md
# MAGIC 
# MAGIC ##### ForeachBatch
# MAGIC 
# MAGIC `foreachBatch(…)`允许指定对流式查询的每个微批的输出数据执行的函数。自Spark 2.4以来，Scala、Java和Python都支持这一点。它有两个参数：一个是DataFrame或Dataset，它包含微批的输出数据，另一个是微批的唯一ID。
# MAGIC 
# MAGIC ```
# MAGIC def foreach_batch_function(df, epoch_id):
# MAGIC     # Transform and write batchDF
# MAGIC     pass
# MAGIC   
# MAGIC streamingDF.writeStream.foreachBatch(foreach_batch_function).start()  
# MAGIC ```
# MAGIC 
# MAGIC 使用foreachBatch，可以执行以下操作。
# MAGIC * 重用现有的批处理数据源 - 对于许多存储系统，可能还没有可用的流式接收器，但可能已经存在用于批处理查询的数据写入程序。使用`foreachBatch`，可以在每个微批的输出上使用批数据写入器。
# MAGIC * 写入多个位置 - 如果想将流式查询的输出写入到多个位置，那么只需多次写入输出DataFrame/Dataset即可。然而，每次写入尝试都会导致重新计算输出数据（包括可能的输入数据的重新读取）。为了避免重新计算，应该缓存输出DataFrame/Dataset，将其写入多个位置，然后取消缓存。
# MAGIC * 应用其他DataFrame操作 - 流式DataFrame中不支持许多DataFrame和Dataset操作，因为Spark不支持在这些情况下生成增量计划。使用`foreachBatch`，可以对每个微批处理输出应用其中的一些操作。但是，必须考虑自己执行该操作的端到端语义。
# MAGIC 
# MAGIC > 注意：  
# MAGIC 默认情况下，foreachBatch只提供至少一次写入保证。但是，可以使用提供给函数的batchId作为消除输出重复的方法，并获得一次保证。
# MAGIC foreachBatch不适用于连续处理模式，因为它基本上依赖于流式查询的微批处理执行。如果以连续模式写入数据，请改用foreach。

# COMMAND ----------

def foreach_batch_function(df, epoch_id):
    print("epoch_id=====>",epoch_id)
    df.printSchema()
    df.show()
    df.write.csv("/mnt/databrickscontainer1/writecsv{}".format(epoch_id))
    df.write.mode("overwrite").format("jdbc").option("url","jdbc:mysql://wux-mysql.mysql.database.azure.com:3306/spark?useSSL=true&requireSSL=false").option("user","wux_labs@wux-mysql").option("password","Pa55w.rd").option("dbtable","spark_stream_write").option("showSql",True).save()
    pass

# bin/kafka-console-producer.sh --bootstrap-server 10.0.0.4:9092 --topic KafkaFirstTopic
df = spark \
  .readStream \
  .format("kafka") \
  .option("kafka.bootstrap.servers", "20.239.80.52:9092") \
  .option("subscribe", "KafkaFirstTopic") \
  .load()

# 对数据做转换、统计
wordCount = df.selectExpr("explode(split(CAST(value AS STRING),\" \"))").groupBy("col").count()

wordCount.writeStream.outputMode("complete").foreachBatch(foreach_batch_function).start()

# COMMAND ----------

# MAGIC %md
# MAGIC 
# MAGIC ##### Foreach
# MAGIC 
# MAGIC 如果`foreachBatch`不是一个选项（例如，对应的批处理数据写入器不存在，或连续处理模式），那么可以使用`foreach`表达自定义写入器逻辑。具体来说，可以通过将数据写入逻辑分为三种方法来表达：`open`、`process`和`close`。自Spark 2.4以来，`foreach`可以在Scala、Java和Python中使用。
# MAGIC 
# MAGIC 在Python中，可以通过两种方式调用foreach：在函数中或在对象中。函数提供了一种简单的方式来表达处理逻辑，但不允许在故障导致某些输入数据重新处理时对生成的数据进行重复数据消除。对于这种情况，必须在对象中指定处理逻辑。
# MAGIC 
# MAGIC * 方式一：函数将一行作为输入
# MAGIC ```
# MAGIC def process_row(row):
# MAGIC     # Write row to storage
# MAGIC     pass
# MAGIC 
# MAGIC query = streamingDF.writeStream.foreach(process_row).start()  
# MAGIC ```
# MAGIC * 方式二：对象有一个`process`方法和可选的`open`和`close`方法
# MAGIC ```
# MAGIC class ForeachWriter:
# MAGIC     def open(self, partition_id, epoch_id):
# MAGIC         # Open connection. This method is optional in Python.
# MAGIC         pass
# MAGIC 
# MAGIC     def process(self, row):
# MAGIC         # Write row to connection. This method is NOT optional in Python.
# MAGIC         pass
# MAGIC 
# MAGIC     def close(self, error):
# MAGIC         # Close the connection. This method in optional in Python.
# MAGIC         pass
# MAGIC       
# MAGIC query = streamingDF.writeStream.foreach(ForeachWriter()).start()
# MAGIC ```
# MAGIC 
# MAGIC 执行语义启动流查询时，Spark以以下方式调用函数或对象的方法：
# MAGIC 
# MAGIC * 此对象的单个副本负责查询中单个任务生成的所有数据。换句话说，一个实例负责处理以分布式方式生成的数据的一个分区。
# MAGIC * 此对象必须是可序列化的，因为每个任务都将获得所提供对象的新的序列化反序列化副本。因此，强烈建议在调用`open()`方法之后完成写入数据的任何初始化（例如，打开连接或启动事务），这意味着任务已准备好生成数据。
# MAGIC * 方法的生命周期如下：
# MAGIC   * 对于具有partition_id的每个分区：
# MAGIC     * 对于具有epoch_id的流数据的每个batch/epoch：
# MAGIC       * 方法`open(partitionId, epochId)`被调用。
# MAGIC       * 如果`open(…)`返回true，对于分区和batch/epoch中的每一行，将调用方法`process(row)`。
# MAGIC       * 在`process(row)`时发现错误（如果有）时，调用方法`close(error)`。
# MAGIC       * 如果`open()`方法存在并成功返回（与返回值无关），则调用`close()`方法（如果存在），除非JVM或Python进程在中间崩溃。
# MAGIC 
# MAGIC > 注意：  
# MAGIC > Spark不保证(partitionId, epochId)的输出相同，因此无法使用(partitionId, epochId)实现重复数据消除。比如，由于某些原因，源提供了不同数量的分区，Spark优化更改了分区数量等。  
# MAGIC > 如果需要对输出进行重复数据消除，请尝试`foreachBatch`。

# COMMAND ----------

def foreach_function(row):
    print("row=====>",type(row),row)
    print(row["col"])
    pass

# bin/kafka-console-producer.sh --bootstrap-server 10.0.0.4:9092 --topic KafkaFirstTopic
df = spark \
  .readStream \
  .format("kafka") \
  .option("kafka.bootstrap.servers", "20.239.80.52:9092") \
  .option("subscribe", "KafkaFirstTopic") \
  .load()

# 对数据做转换、统计
wordCount = df.selectExpr("explode(split(CAST(value AS STRING),\" \"))").groupBy("col").count()

wordCount.writeStream.outputMode("complete").foreach(foreach_function).start()

# COMMAND ----------

# MAGIC %sh
# MAGIC 
# MAGIC # pip install pymysql -i https://pypi.douban.com/simple
# MAGIC pip install pymysql -i https://pypi.tuna.tsinghua.edu.cn/simple
# MAGIC 
# MAGIC pip list

# COMMAND ----------

import pymysql

db = pymysql.connect(host="wux-mysql.mysql.database.azure.com",user="wux_labs@wux-mysql",password="Pa55w.rd",database="spark")
cursor = db.cursor()
cursor.execute("insert into spark_stream_write(col,count) values('replcace',1) ON DUPLICATE KEY UPDATE count = 1")
cursor.execute("select * from spark_stream_write")

# COMMAND ----------

import pymysql
from pymysql.connections import Connection

class ForeachWriterLabs:
    
    db: Connection = None
    
    def open(self, partition_id, epoch_id):
        # Open connection. This method is optional in Python.
        print("Open connection. =====> This method is optional in Python.")
        self.db = pymysql.connect(host="wux-mysql.mysql.database.azure.com",user="wux_labs@wux-mysql",password="Pa55w.rd",database="spark",autocommit=True)
        return True

    def process(self, row):
        sql = "insert into spark_stream_write(col,count) values('{}',{}) ON DUPLICATE KEY UPDATE count = {}".format(row["col"],str(row["count"]),str(row["count"]))
        self.db.cursor().execute(sql)
        pass

    def close(self, error):
        # Close the connection. This method in optional in Python.
        print("Close the connection. =====> This method is optional in Python.")
        # self.db.commit()
        self.db.close()
        pass

# bin/kafka-console-producer.sh --bootstrap-server 10.0.0.4:9092 --topic KafkaFirstTopic
df = spark \
  .readStream \
  .format("kafka") \
  .option("kafka.bootstrap.servers", "20.239.80.52:9092") \
  .option("subscribe", "KafkaFirstTopic") \
  .load()

# 对数据做转换、统计
wordCount = df.selectExpr("explode(split(CAST(value AS STRING),\" \"))").groupBy("col").count()

wordCount.writeStream.outputMode("complete").foreach(ForeachWriterLabs()).start()

# COMMAND ----------

# MAGIC %md
# MAGIC 
# MAGIC **注意：** 如果需要创建数据源连接：
# MAGIC * 不要在Driver端创建连接，因为可能出现序列化反序列化失败的问题
# MAGIC * 不要为每个RDD的数据创建一个连接，一方面会导致连接数过多，一方面创建连接的开销很大
# MAGIC * 尽量让一次连接处理一个分区的数据
# MAGIC * 可以使用共享连接池

# COMMAND ----------

# MAGIC %md
# MAGIC 
# MAGIC ### 流式处理表API
# MAGIC 
# MAGIC 从Spark 3.1开始，可以使用DataStreamReader.table()将表作为流DataFrames读取，并使用DataStreamWriter.toTable()将流DataFrames写入表：
# MAGIC ```
# MAGIC spark = ...  # spark session
# MAGIC 
# MAGIC # Create a streaming DataFrame
# MAGIC df = spark.readStream \
# MAGIC     .format("rate") \
# MAGIC     .option("rowsPerSecond", 10) \
# MAGIC     .load()
# MAGIC 
# MAGIC # Write the streaming DataFrame to a table
# MAGIC df.writeStream \
# MAGIC     .option("checkpointLocation", "path/to/checkpoint/dir") \
# MAGIC     .toTable("myTable")
# MAGIC 
# MAGIC # Check the table result
# MAGIC spark.read.table("myTable").show()
# MAGIC 
# MAGIC # Transform the source dataset and write to a new table
# MAGIC spark.readStream \
# MAGIC     .table("myTable") \
# MAGIC     .select("value") \
# MAGIC     .writeStream \
# MAGIC     .option("checkpointLocation", "path/to/checkpoint/dir") \
# MAGIC     .format("parquet") \
# MAGIC     .toTable("newTable")
# MAGIC 
# MAGIC # Check the new table result
# MAGIC spark.read.table("newTable").show()
# MAGIC ```

# COMMAND ----------

df = spark \
  .readStream \
  .format("kafka") \
  .option("kafka.bootstrap.servers", "20.239.80.52:9092") \
  .option("subscribe", "KafkaFirstTopic") \
  .load()

# 对数据做转换、统计
wordCount = df.selectExpr("explode(split(CAST(value AS STRING),\" \"))").groupBy("col").count()

wordCount.writeStream.outputMode("complete") \
    .option("checkpointLocation", "/mnt/databrickscontainer1/checkpoint_table") \
    .toTable("wordCountTable")

# 读取的是静态数据
spark.read.table("wordCountTable").show()

# 读取的是流式数据
# Detected a data update (for example part-00047-4e8f00c1-a56b-44c7-8045-b77b8f94c2ad-c000.snappy.parquet) in the source table at version 5. This is currently not supported.
display(spark.readStream.table("wordCountTable"))

# COMMAND ----------

# MAGIC %md
# MAGIC 
# MAGIC ### 触发器
# MAGIC 
# MAGIC 流式查询的触发设置定义了流式数据处理的定时，即查询是将以固定的批处理间隔作为微批处理查询执行还是作为连续处理查询执行。下面是支持的不同类型的触发器。
# MAGIC 
# MAGIC 
# MAGIC | 触发器类型 | 描述 |
# MAGIC | :- | :- |
# MAGIC | 未指定（默认） | 如果未显式指定触发器设置，则默认情况下，查询将以微批处理模式执行，其中微批处理将在前一个微批处理已完成时进行处理。 |
# MAGIC | 固定间隔微批次 | 查询将以微批处理模式执行，其中微批处理将以用户指定的时间间隔启动。<li>如果上一个微批处理在间隔内完成，则引擎将等到上一个时间间隔结束，再开始下一个微批处理。</li><li>如果前一个微批处理需要比间隔更长的时间才能完成（即，一个微批时间间隔无法保证当前微批的数据处理完成），那么下一个微批处理将在前一个微批完成时立即开始（即，它不会等待下一个时间间隔）。</li><li>如果没有新数据可用，则不会启动微批处理。</li> |
# MAGIC | 一次性微批量 | 查询将**仅执行一个**微批处理来处理所有可用数据，然后自行停止。这在要定期启动群集的方案中很有用，处理自上一个周期以来可用的所有内容，然后关闭。在某些情况下，这可能会节省大量成本。 |
# MAGIC | 具有固定检查点间隔的连续性（实验） | 查询将在新的低延迟连续处理模式下执行。 |
# MAGIC 
# MAGIC ```
# MAGIC # Default trigger (runs micro-batch as soon as it can)
# MAGIC df.writeStream \
# MAGIC   .format("console") \
# MAGIC   .start()
# MAGIC 
# MAGIC # ProcessingTime trigger with two-seconds micro-batch interval
# MAGIC df.writeStream \
# MAGIC   .format("console") \
# MAGIC   .trigger(processingTime='2 seconds') \
# MAGIC   .start()
# MAGIC 
# MAGIC # One-time trigger
# MAGIC df.writeStream \
# MAGIC   .format("console") \
# MAGIC   .trigger(once=True) \
# MAGIC   .start()
# MAGIC 
# MAGIC # Continuous trigger with one-second checkpointing interval
# MAGIC df.writeStream
# MAGIC   .format("console")
# MAGIC   .trigger(continuous='1 second')
# MAGIC   .start()
# MAGIC   ```

# COMMAND ----------

df = spark \
  .readStream \
  .format("kafka") \
  .option("kafka.bootstrap.servers", "104.208.119.0:9092") \
  .option("subscribe", "KafkaFirstTopic") \
  .load()

# 原始数据
df.writeStream.format("console").start()

# COMMAND ----------

df = spark \
  .readStream \
  .format("kafka") \
  .option("kafka.bootstrap.servers", "104.208.119.0:9092") \
  .option("subscribe", "KafkaFirstTopic") \
  .trigger(processingTime='2 minutes') \
  .load()

# 原始数据
df.writeStream.format("console").start()

# COMMAND ----------

df = spark \
  .readStream \
  .format("kafka") \
  .option("kafka.bootstrap.servers", "104.208.119.0:9092") \
  .option("subscribe", "KafkaFirstTopic") \
  .trigger(once=True) \
  .load()

# 原始数据
df.writeStream.format("console").start()

# COMMAND ----------

# MAGIC %md
# MAGIC 
# MAGIC ## 管理流式处理查询
# MAGIC 
# MAGIC 启动查询时创建的StreamingQuery对象可用于监视和管理查询。
# MAGIC 
# MAGIC ```
# MAGIC query = df.writeStream.format("console").start()   # get the query object
# MAGIC 
# MAGIC query.id()          # get the unique identifier of the running query that persists across restarts from checkpoint data
# MAGIC 
# MAGIC query.runId()       # get the unique id of this run of the query, which will be generated at every start/restart
# MAGIC 
# MAGIC query.name()        # get the name of the auto-generated or user-specified name
# MAGIC 
# MAGIC query.explain()   # print detailed explanations of the query
# MAGIC 
# MAGIC query.stop()      # stop the query
# MAGIC 
# MAGIC query.awaitTermination()   # block until query is terminated, with stop() or with error
# MAGIC 
# MAGIC query.exception()       # the exception if the query has been terminated with error
# MAGIC 
# MAGIC query.recentProgress  # a list of the most recent progress updates for this query
# MAGIC 
# MAGIC query.lastProgress    # the most recent progress update of this streaming query
# MAGIC ```
# MAGIC 
# MAGIC 可以在一个SparkSession中启动任意数量的查询。它们都将同时运行，共享集群资源。可以使用sparkSession.streams()获取StreamingQueryManager，可用于管理当前活动的查询。
# MAGIC 
# MAGIC ```
# MAGIC spark = ...  # spark session
# MAGIC 
# MAGIC spark.streams.active  # get the list of currently active streaming queries
# MAGIC 
# MAGIC spark.streams.get(id)  # get a query object by its unique id
# MAGIC 
# MAGIC spark.streams.awaitAnyTermination()  # block until any one of them terminates
# MAGIC ```

# COMMAND ----------

df = spark \
  .readStream \
  .format("kafka") \
  .option("kafka.bootstrap.servers", "20.239.80.52:9092") \
  .option("subscribe", "KafkaFirstTopic") \
  .load()

# 原始数据
query = df.writeStream.format("console").start()

print("id=====>", query.id)
print("lastProgress=====>", query.lastProgress)

query.awaitTermination()

# COMMAND ----------

# MAGIC %md
# MAGIC 
# MAGIC ## 监视流式处理查询
# MAGIC 
# MAGIC 有多种方法可以监视活动的流查询。您可以使用Spark的Dropwizard metrics支持将metrics推送到外部系统，也可以通过编程方式访问它们。

# COMMAND ----------

# MAGIC %md
# MAGIC 
# MAGIC ### 以交互方式读取指标
# MAGIC 
# MAGIC 可以使用streamingQuery.lastProgress()和streamingQuery.status()直接获取活动查询的当前状态和度量。lastProgress()在Scala和Java中返回StreamingQueryProgress对象，在Python中返回具有相同字段的字典。它包含流的最后一个触发器中的所有进度信息-处理了哪些数据、处理率、延迟等。还有streamingQuery.recentProgress，返回最后几个进度的数组。
# MAGIC 
# MAGIC 此外，streamingQuery.status()在Scala和Java中返回StreamingQueryStatus对象，在Python中返回具有相同字段的字典。它提供了有关查询立即执行的操作的信息-触发器是否处于活动状态，数据是否正在处理等。
# MAGIC 
# MAGIC ```
# MAGIC query = ...  # a StreamingQuery
# MAGIC print(query.lastProgress)
# MAGIC 
# MAGIC '''
# MAGIC Will print something like the following.
# MAGIC 
# MAGIC {u'stateOperators': [], u'eventTime': {u'watermark': u'2016-12-14T18:45:24.873Z'}, u'name': u'MyQuery', u'timestamp': u'2016-12-14T18:45:24.873Z', u'processedRowsPerSecond': 200.0, u'inputRowsPerSecond': 120.0, u'numInputRows': 10, u'sources': [{u'description': u'KafkaSource[Subscribe[topic-0]]', u'endOffset': {u'topic-0': {u'1': 134, u'0': 534, u'3': 21, u'2': 0, u'4': 115}}, u'processedRowsPerSecond': 200.0, u'inputRowsPerSecond': 120.0, u'numInputRows': 10, u'startOffset': {u'topic-0': {u'1': 1, u'0': 1, u'3': 1, u'2': 0, u'4': 1}}}], u'durationMs': {u'getOffset': 2, u'triggerExecution': 3}, u'runId': u'88e2ff94-ede0-45a8-b687-6316fbef529a', u'id': u'ce011fdc-8762-4dcb-84eb-a77333e28109', u'sink': {u'description': u'MemorySink'}}
# MAGIC '''
# MAGIC 
# MAGIC print(query.status)
# MAGIC ''' 
# MAGIC Will print something like the following.
# MAGIC 
# MAGIC {u'message': u'Waiting for data to arrive', u'isTriggerActive': False, u'isDataAvailable': False}
# MAGIC '''
# MAGIC ```

# COMMAND ----------

df = spark \
  .readStream \
  .format("kafka") \
  .option("kafka.bootstrap.servers", "20.239.80.52:9092") \
  .option("subscribe", "KafkaFirstTopic") \
  .load()

# 原始数据
query = df.writeStream.format("console").start()

print("lastProgress=====>", query.lastProgress)
print("status=====>", query.status)

# COMMAND ----------

# MAGIC %md
# MAGIC 
# MAGIC ### 使用异步API以编程方式报告指标
# MAGIC 
# MAGIC 还可以通过附加StreamingQueryListener（Scala/Java文档）异步监视与SparkSession关联的所有查询。将自定义StreamingQueryListener对象附加到sparkSession.streams之后。attachListener()，当查询启动和停止以及活动查询中有进展时，将得到回调。
# MAGIC 
# MAGIC > 注意：Python不支持这种方式。  
# MAGIC > [https://spark.apache.org/docs/3.2.1/structured-streaming-programming-guide.html#reporting-metrics-programmatically-using-asynchronous-apis](https://spark.apache.org/docs/3.2.1/structured-streaming-programming-guide.html#reporting-metrics-programmatically-using-asynchronous-apis)
# MAGIC 
# MAGIC > StreamingQueryListener在Scala中是一个抽象类。  
# MAGIC > [https://github.com/apache/spark/blob/master/sql/core/src/main/scala/org/apache/spark/sql/streaming/StreamingQueryListener.scala](https://github.com/apache/spark/blob/master/sql/core/src/main/scala/org/apache/spark/sql/streaming/StreamingQueryListener.scala)
# MAGIC 
# MAGIC [monitor](https://www.databricks.com/blog/2022/05/27/how-to-monitor-streaming-queries-in-pyspark.html)

# COMMAND ----------

# MAGIC %scala
# MAGIC 
# MAGIC import org.apache.spark.sql.streaming.StreamingQueryListener
# MAGIC import org.apache.spark.sql.streaming.StreamingQueryListener._
# MAGIC import org.apache.spark.streaming.scheduler.StreamingListener
# MAGIC import spark.implicits._
# MAGIC 
# MAGIC val myListener = new StreamingQueryListener() {
# MAGIC     override def onQueryStarted(queryStarted: QueryStartedEvent): Unit = {
# MAGIC         println("Query started: " + queryStarted.id)
# MAGIC     }
# MAGIC     override def onQueryTerminated(queryTerminated: QueryTerminatedEvent): Unit = {
# MAGIC         println("Query terminated: " + queryTerminated.id)
# MAGIC     }
# MAGIC     override def onQueryProgress(queryProgress: QueryProgressEvent): Unit = {
# MAGIC         println("Query made progress: " + queryProgress.progress)
# MAGIC     }
# MAGIC }
# MAGIC 
# MAGIC spark.streams.addListener(myListener)
# MAGIC 
# MAGIC val df = spark
# MAGIC   .readStream
# MAGIC   .format("kafka")
# MAGIC   .option("kafka.bootstrap.servers", "20.239.80.52:9092")
# MAGIC   .option("subscribe", "KafkaFirstTopic")
# MAGIC   .load()
# MAGIC 
# MAGIC // 原始数据
# MAGIC val query = df.writeStream.format("console").start()
# MAGIC 
# MAGIC println("id=====>", query.id)
# MAGIC println("lastProgress=====>", query.lastProgress)
# MAGIC 
# MAGIC // query.awaitTermination()

# COMMAND ----------

# MAGIC %scala
# MAGIC 
# MAGIC spark.streams.removeListener(myListener)

# COMMAND ----------

from pyspark.sql.streaming import StreamingQueryListener

class MyListener(StreamingQueryListener):
    def onQueryStarted(self, event):
        """
        Called when a query is started.

        Parameters
        ----------
        event: :class:`pyspark.sql.streaming.listener.QueryStartedEvent`
            The properties are available as the same as Scala API.

        Notes
        -----
        This is called synchronously with
        meth:`pyspark.sql.streaming.DataStreamWriter.start`,
        that is, ``onQueryStart`` will be called on all listeners before
        ``DataStreamWriter.start()`` returns the corresponding
        :class:`pyspark.sql.streaming.StreamingQuery`.
        Do not block in this method as it will block your query.
        """
        pass

    def onQueryProgress(self, event):
        """
        Called when there is some status update (ingestion rate updated, etc.)

        Parameters
        ----------
        event: :class:`pyspark.sql.streaming.listener.QueryProgressEvent`
            The properties are available as the same as Scala API.

        Notes
        -----
        This method is asynchronous. The status in
        :class:`pyspark.sql.streaming.StreamingQuery` will always be
        latest no matter when this method is called. Therefore, the status
        of :class:`pyspark.sql.streaming.StreamingQuery`.
        may be changed before/when you process the event.
        For example, you may find :class:`StreamingQuery`
        is terminated when you are processing `QueryProgressEvent`.
        """
        pass

    def onQueryTerminated(self, event):
        """
        Called when a query is stopped, with or without error.

        Parameters
        ----------
        event: :class:`pyspark.sql.streaming.listener.QueryTerminatedEvent`
            The properties are available as the same as Scala API.
        """
        pass


my_listener = MyListener()

# COMMAND ----------

from pyspark.sql.streaming import StreamingQueryListener

class MyListener(StreamingQueryListener):
    def onQueryStarted(self, event):
        print("a query is started.")
        pass

    def onQueryProgress(self, event):
        print("event=====>", event)
        print("event.progress.id=====>", event.progress.id)
        print("event.progress=====>", event.progress)
        pass

    def onQueryTerminated(self, event):
        print("a query is stopped.")
        pass

my_listener = MyListener()

# COMMAND ----------

spark.streams.addListener(my_listener)

# COMMAND ----------

df = spark \
  .readStream \
  .format("kafka") \
  .option("kafka.bootstrap.servers", "20.239.80.52:9092") \
  .option("subscribe", "KafkaFirstTopic") \
  .load()

# 原始数据
query = df.writeStream.format("console").start()

print("id=====>", query.id)
print("lastProgress=====>", query.lastProgress)


# COMMAND ----------

spark.streams.removeListener(my_listener)

# COMMAND ----------

# MAGIC %md
# MAGIC 
# MAGIC ### 使用Dropwizard报告指标
# MAGIC 
# MAGIC Spark支持使用Dropwizard库报告指标。为了使结构化流查询的指标也能够被报告，必须显式地在SparkSession中启用配置`spark.sql.Streaming.metricsEnabled`。
# MAGIC 
# MAGIC ```
# MAGIC spark.conf.set("spark.sql.streaming.metricsEnabled", "true")
# MAGIC # or
# MAGIC spark.sql("SET spark.sql.streaming.metricsEnabled=true")
# MAGIC ```
